package com.turman.coroutionsample

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.sync.withPermit
import org.junit.Test
import java.io.BufferedReader
import java.io.FileReader
import java.io.IOException
import java.util.concurrent.atomic.AtomicInteger
import kotlin.system.measureTimeMillis

class CoroutineTest01 {
    @Test
    fun `test coroutine builder`() = runBlocking {
        val job1 = launch {
            delay(200)
            println("launch finished.")
        }
        val job2 = async {
            delay(200)
            println("async finished.")
            "job2 result"
        }
        println(job2.await())
    }

    @Test
    fun `test coroutine join`() = runBlocking {
        val job1 = launch {
            delay(300)
            println("job one")
        }
        job1.join()  //添加join可以保证job1执行完后再执行后面的任务，这里输出顺序是 one three two
        val job2 = launch {
            delay(200)
            println("job two")
        }
//        job2.join()
        val job3 = launch {
            delay(100)
            println("job three")
        }
//        job3.join()
    }

    @Test
    fun `test coroutine await`() = runBlocking {
        val job1 = async {
            delay(300)
            println("job one")
        }
        job1.await() //job1会执行完后再去执行其他job
        val job2 = async {
            delay(200)
            println("job two")
        }
        val job3 = async {
            delay(100)
            println("job three")
        }
    }

    @Test
    fun `test coroutine ca`() = runBlocking {
        val time = measureTimeMillis {
            val num1 = getOne()
            val num2 = getTwo()
            println("reasult: ${num1 + num2}")
        }
        println("time:${time} ms")
    }

    @Test
    fun `test coroutine launch`() = runBlocking {
        val time = measureTimeMillis {
            val job1 = launch { getOne() }
            val job2 = launch { getTwo() }
            //和上面例子的区别，delay没有写在协程体里面 而是卸载了函数调用里面，所以编译器检测不到delay就需要调用join来等待执行了？
            joinAll(job1, job2) //这里需要添加join才能执行完 ，为什么？
        }
        println("time:${time} ms")
    }

    @Test
    fun `test coroutine sync`() = runBlocking {
        /**
         * 通过上面的例子个这个例子总结下：
         * launch和async都会开启协程，并发执行。
         * 当不指定launch和async时是在当前协程顺序执行的
         */
        val time = measureTimeMillis {
            val num1 = async { getOne() }
            val num2 = async { getTwo() }
            println("reasult: ${num1.await() + num2.await()}")
        }
        println("time:${time} ms")
    }

    @Test
    fun `test coroutine sync wrong`() = runBlocking {
        val time = measureTimeMillis {
            /**
             * 不太聪明的调用方式
             */
            val num1 = async { getOne() }.await()
            val num2 = async { getTwo() }.await()
            println("reasult: ${num1 + num2}")
        }
        println("time:${time} ms")
    }

    private suspend fun getOne(): Int {
        delay(1000)
        return 2
    }

    private suspend fun getTwo(): Int {
        delay(1200)
        return 4
    }

    @Test
    fun `test start mode 1`() = runBlocking {
        val job = launch(start = CoroutineStart.DEFAULT) {
            try {
                delay(3000)
                println("coroutine finished.")
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        delay(1000)
        job.cancel()
    }

    @Test
    fun `test start mode 2`() = runBlocking {
        val job = launch(start = CoroutineStart.LAZY) {
            try {
                delay(3000)
                println("coroutine finished.")
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        /**
         * 这种启动模式下只有调用start才会执行协程。所以适合预定于一些协程以待后续执行
         */
        job.start() //join await也会触发执行
        delay(1000)
        job.cancel()
    }

    @Test
    fun `test start mode 3`() = runBlocking {
        /**
         * 比较这两个job得出，ATOMIC方式启动协程时，协程在执行到第一个挂起点前不会退出，
         * 这是它的特性
         */
        val job1 = launch(start = CoroutineStart.ATOMIC) {
            try {
                repeat(1000) {
                    println("job1 $it")
                    if (it == 200) {
                        delay(1000)
                    }
                }
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        job1.cancel()

        val job2 = launch(start = CoroutineStart.ATOMIC) {
            try {
                repeat(1000) {
                    println("job2 $it")
                }
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        job2.cancel()
    }

    @Test
    fun `test start mode 4`() = runBlocking {
        /**
         * 协程创建后立即在当前函数栈上执行，直到第一个挂起函数
         * 所以协程在执行到delay前都是在当前协程所在的线程执行，delay之后会通过io线程持中的线程执行
         */
        val job = launch(context = Dispatchers.IO, start = CoroutineStart.UNDISPATCHED) {
            println("thread:${Thread.currentThread().name}")
            delay(1000)
            println("thread:${Thread.currentThread().name}")
        }
    }

    @Test
    fun `test coroutine scope builder 1`() = runBlocking {
        /**
         * coroutineScope这个协程作用域会保证被其包含的所有协程都执行完成
         * 如果有一个协程执行异常，则该作用域写的其他协程也会退出
         */
        val result = coroutineScope {
            val job1 = launch {
                delay(400)
                println("job1 finished")
            }
            val job2 = async {
                delay(200)
                println("job2 finished")
                "job2 result"
                throw IllegalArgumentException()  //这里抛出异常会导致job1退出执行，从而退出协程作用域
            }

            "${job2.await()}, all job finished"
        }
        println(result)
    }

    @Test
    fun `test coroutine scope builder 2`() = runBlocking {
        /**
         * supervisorScope这个协程作用域会保证被其包含的所有协程互不干扰
         * 如果有一个协程执行异常，则其他协程仍然能执行完成
         */
        supervisorScope {
            val job1 = launch {
                delay(400)
                println("job1 finished")
            }
            val job2 = async {
                delay(200)
                println("job2 finished")
                "job2 result"
                throw IllegalArgumentException()  //这里抛出异常会导致job1退出执行，从而退出协程作用域
            }

//            "${job2.await()}, all job finished"
        }
//        println(result)
    }

    @Test
    fun `test scope cancel 1`() = runBlocking {
        //创建一个协程作用域
        val scope = CoroutineScope(Dispatchers.Default)

        scope.launch {
            delay(1000)
            println("job 1 finished.")
        }

        scope.launch {
            delay(2000)
            println("job 2 finished.")
        }
        scope.launch {
            delay(1500)
            println("job 3 finished.")
        }
        delay(1000)
        scope.cancel() //只有job1能执行完成，job2和job3都会被取消
    }

    @Test
    fun `test scope cancel 2`() = runBlocking {
        val scope = CoroutineScope(Dispatchers.Default)

        val job1 = scope.launch {
            delay(1000)
            println("job 1 finished.")
        }

        val job2 = scope.launch {
            delay(2000)
            println("job 2 finished.")
        }
        val job3 = scope.launch {
            delay(1500)
            println("job 3 finished.")
        }
        delay(1000)
        job3.cancel() //子协程取消不会影响其他兄弟协程执行

        joinAll(job1, job2, job3)
    }

    @Test
    fun `test scope cancel exception`() = runBlocking {
        val job1 = GlobalScope.launch {
            try {
                delay(1000)
                println("job 1")
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        delay(200)
        job1.cancel(CancellationException("取消")) //job取消时可以指定取消异常参数
        job1.join()
//        job1.cancelAndJoin()
    }

    @Test
    fun `test cancel cpu task by isActive`() = runBlocking {
        /**
         * 计算密集型任务无法取消
         */
        val startTime = System.currentTimeMillis()
        val job = launch(Dispatchers.Default) {
            var nextPrintTime = startTime
            var i = 0
            while (i < 5) { //如果需要这个计算密集型任务被取消逍遥添加这个条件 while(i < 5 && isActive)
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job:I'm sleeping ${i++}...")
                    nextPrintTime += 500
                }
            }
        }
        println("main:print over 1")
        job.cancel()

        //这个任务可以被取消
        val startTime1 = System.currentTimeMillis()
        val job1 = launch {
            var nextPrintTime = startTime1
            var i = 0
            while (i < 5) {
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job:I'm sleepting ${i++}...")
                    nextPrintTime += 500
                }
            }
        }
        println("main:print over 2")
        job1.cancel()
    }

    @Test
    fun `test cancel cpu task by ensureActive`() = runBlocking {
        val startTime = System.currentTimeMillis()
        val job = launch(Dispatchers.Default) {
            var nextPrintTime = startTime
            var i = 0
            while (i < 5) {
                ensureActive()  //添加这个可以检测取消操作以取消协程
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job:I'm sleeping ${i++}...")
                    nextPrintTime += 500
                }
            }
        }
        job.cancel()
    }

    @Test
    fun `test cancel cpu task by yield`() = runBlocking {
        //通过调用yield使协程主动让出执行权 可以使计算密集型协程及时取消
        val startTime = System.currentTimeMillis()
        val job = launch(Dispatchers.Default) {
            var nextPrintTime = startTime
            var i = 0
            while (i < 5) {
                yield() //yield的意思是让出当前协程的执行权，如果当前只有这个协程在执行，则会继续执行这个协程，此时取消会得到响应并退出协程执行
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job:I'm sleeping ${i++}...")
                    nextPrintTime += 500
                }
            }
        }
        delay(1000)
        job.cancel()
    }

    @Test
    fun `test cancel release resources`() = runBlocking {
        val job = launch {
            try {
                repeat(1000) {
                    println("this is $it times print.")
                    delay(500)
                }
            } catch (e: Exception) {
                e.printStackTrace()
            } finally {
                //取消协程时 需要释放的资源可以放在这里执行
                println("resources release.")
            }
        }
        delay(2000)
        job.cancel()
    }

    @Test
    fun `test cancel use function`() = runBlocking {
        val job = launch {
            var br =
                BufferedReader(FileReader("D:\\project\\CoroutionSample\\app\\src\\test\\java\\com\\turman\\coroutionsample\\test.txt"))

//            with(br) {
//                var line: String?
//                while (true) {
//                    line = readLine() ?: break
//                    println(line)
//                    delay(200)
//                }
//                close()
//            }

            br.use { //通过use关闭资源
                var line: String?
                while (true) {
                    line = readLine() ?: break
                    println(line)
                }
            }
        }
        delay(1000)
        job.cancel()
    }

    @Test
    fun `test cancel with NonCancellable`() = runBlocking {
        val job = launch {
            try {
                repeat(1000) {
                    println("job:I'm sleeping $it ...")
                    delay(500)
                }
            } catch (e: Exception) {
                e.printStackTrace()
            } finally {
//                println("finally print")
//                delay(1000) //finally中调用delay 之后的内容不会执行
//                println("finally print after delay")

                withContext(NonCancellable) {
                    delay(2000)
                    println("finally print in NonCancellable")
                }
            }
        }
        delay(1000)
        job.cancel()
    }

    @Test
    fun `test deal with timeout`() = runBlocking {
        /**
         * 关于超时任务的处理
         */
        val job1 = launch {
            val result = withTimeout(1000) {
                repeat(1000) {
                    println("print1 $it times.")
                    delay(500)
                }
                "Done"
            }
            println("result1:$result")
        }

        val job2 = launch {
            val result = withTimeoutOrNull(1000) {
                repeat(10) {
                    println("print2 $it times.")
                    delay(500)
                }
                "Done"
            } ?: "TimeOut"
            println("result2:$result")
        }

        joinAll(job1, job2)
    }

    @Test
    fun `test coroutine context`() = runBlocking<Unit> {
        launch(Dispatchers.Default + CoroutineName("buobao-text")) {
            println("Thread:${Thread.currentThread().name} ${coroutineContext[CoroutineName]}")
        }
    }

    @Test
    fun `test coroutine context extend 1`() = runBlocking<Unit> {
        /**
         * 对于新创建的协程，它的CoroutineContext会包含一个全新的Job实例，它会帮助我们控制协程的生命周期。而剩下的元素会从CoroutineContext的父类继承，
         * 该父类可能是另一个协程或者创建该协程的CoroutineContext.
         */
        val scope = CoroutineScope(Job() + Dispatchers.IO + CoroutineName("buobao-test"))
        val job = scope.launch {
            println("${coroutineContext[Job]} ${Thread.currentThread().name}")
            val result = async {
                println("${coroutineContext[Job]} ${Thread.currentThread().name}")
                "OK"
            }.await()
        }
        job.join()
    }

    @Test
    fun `test coroutine context extend 2`() = runBlocking<Unit> {
        /**
         * 指定异常处理器
         */
        val coroutineExceptionHandler = CoroutineExceptionHandler { _, exception ->
            println("Caught $exception")
        }
        val scope = CoroutineScope(Job() + Dispatchers.Main + coroutineExceptionHandler)
        val job = scope.launch(Dispatchers.IO) {

        }
    }

    @Test
    fun `test exception catch`() = runBlocking<Unit> {
        /**
         * launch和async在不同的地方获取异常
         */
        val job = GlobalScope.launch {
            try {
                throw IllegalArgumentException()
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
        val deferred = GlobalScope.async {
            throw NullPointerException()
        }

        try {
            deferred.await()
        } catch (e: Exception) {
            e.printStackTrace()
        }

        joinAll(job, deferred)
    }

    @Test
    fun `test exception supervisorJob`() = runBlocking {
        //一个协程异常保证兄弟协程和父协程正常执行
        /**
         * 见证supervisorJob 和 Job 的区别
         */
        val supervisorScope = CoroutineScope(SupervisorJob())
        val job1 = supervisorScope.launch {
            delay(100)
            println("job 1")
            throw IllegalArgumentException()
        }

        val job2 = supervisorScope.launch {
            delay(300)
            println("job 2")
        }

        joinAll(job1, job2)
    }

    @Test
    fun `test coroutineExceptionHandler 1`() {
        /**
         * launch异常可以捕获，async异常没有捕获
         */

        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught $exception")
        }

        val job = GlobalScope.launch(handler) {
            delay(100)
            println("job finished.")
            throw AssertionError()
        }

        val deferred = GlobalScope.async(handler) {
            println("deferred finished.")
            throw ArithmeticException()
        }

        Thread.sleep(1000)

//        job.join()
//        deferred.await()
    }

    @Test
    fun `test coroutineExceptionHandler 2`() = runBlocking<Unit> {
        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught $exception")
        }

        val scope = CoroutineScope(SupervisorJob())
        val job = scope.launch() {
            async(Dispatchers.IO) {
                delay(10)
                throw IllegalArgumentException()
            }
            launch(handler) {
                println("launch finished.") //异常处理器要放在根协程或者协程作用域上才能起作用
                throw java.lang.NullPointerException()
            }
        }

        job.join()
    }

    @Test
    fun `test cancel exception sending`() = runBlocking {
        /**
         * 取消异常不会传递给父协程
         */
        val job = launch {
            val child = launch {
                try {
                    delay(2000)
                } finally {
                    println("child canceled.")
                }
            }
            yield()
            println("canceling child")
            child.cancelAndJoin()
            yield()
            println("parent is not canceled")
        }
        job.join()
    }

    @Test
    fun `test exception flow`() = runBlocking {
        /**
         * 协程异常处理顺序，
         * 异常发生后首先通知其他兄弟协程取消，等待他们取消完成后，再有异常处理器处理异常。
         */
        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught: $exception")
        }

        val job = GlobalScope.launch(handler) {
            launch {
                try {
                    println("job 1 start.")
                    delay(Long.MAX_VALUE)
                } finally {
                    withContext(NonCancellable) {
                        println("job 1 start cancel.")
                        delay(100)
                        println("job 1 cancel finished.")
                    }
                }
            }

            launch {
                println("job 2 start.")
                throw java.lang.NullPointerException()
            }
        }
        job.join()
    }

    @Test
    fun `test exception aggregation`() = runBlocking {
        /**
         * 异常的聚合 一个异常引发另一个异常 在捕获的时候如何获取？
         */
        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught: $exception ${exception.suppressed.contentToString()}")
        }

        val job = GlobalScope.launch(handler) {
            launch {
                try {
                    delay(Long.MAX_VALUE)
                } finally {
                    throw java.lang.ArithmeticException()  //第二个异常
                }
            }

            launch {
                try {
                    delay(Long.MAX_VALUE)
                } finally {
                    throw OutOfMemoryError()  //第三个异常
                }
            }

            launch {
                delay(100)
                throw IOException()  //第一个异常
            }
        }
        job.join()
    }


    /**
     * flow-异步的返回多个值 怎么做？他可以应用在什么开发场景？
     * 比如：要通过网络获取用户列表的信息，信息中同时有用户的头像和等级，但是这两个数据是通过其他接口获取的
     */
    @Test
    fun `test request mutidata`() = runBlocking {
        //同步获取多个值可以直接返回数组，如：
        val list = getUserList1()

        list.forEach {
            println(it)
        }
        //如果数组中的数据是异步获取的呢？
        val time = measureTimeMillis {
            //这里返回的序列顺序是固定的 显然第一个元素获取的时间最长，他是等待多有元素都获取后才返回整个列表
            val netList = getNetUser()
            netList.forEach {
                println(it)
            }
        }
        println("time:$time")

        //以上获取大批量异步数据就不合适了，我不能起很多协程去生产吧？
        //sequence实现
        getSequenceUser().forEach {
            println(it)
        }

        /**
         *         //如果要处理这样一种场景：
         *         //我们需要获取很多数据 这些数据由不同的接口获取，
         *         //每个接口获取数据的时间不定，当我们在下游处理的时候依据数据返回的先后顺序一次处理
         *         //这时候就需要用到flow了
         *
         */
        getUserFlow().collect { //下游通过collect来接受数据
            println(it)
        }

    }

    private fun getUserList1() = listOf("Jack", "James", "Tom")

    private suspend fun getNetUser() = coroutineScope {
        val job1 = async {
            delay(500)
            "net-Jack"
        }
        val job2 = async {
            delay(200)
            "net-James"
        }
        val job3 = async {
            delay(300)
            "net-Tom"
        }
        listOf(job1.await(), job2.await(), job3.await())
    }

    private fun getSequenceUser(): Sequence<String> = sequence {
        val list = listOf<String>("seq-Jack", "seq-James", "seq-Tom")
        for (i in 0 until 3) {
//            delay(100)
            Thread.sleep(1000) //这里是阻塞线程的方式，效率低 不推荐，这里模拟一种异步获取的场景
            yield(list[i])
        }
    }

    private fun getUserFlow(): Flow<String> = flow {
        val list = listOf("flow-Jack", "flow-James", "flow-Tome")
        for (i in 0 until 3) {
            delay(1000)
            emit(list[i])
        }
    }

    @Test
    fun `test flow is cold`() = runBlocking {
        /**
         * 冷流的意思
         */
        val flow = simpleFlow()
        println("collect start...")
        flow.collect {
            println(it)
        }
        println("collect start again...")
        flow.collect {
            println(it)
        }
    }

    private fun simpleFlow() = flow {
        println("start flow...")
        for (i in 1..5) {
            delay(1000)
            emit(i)
        }
    }

    @Test
    fun `test flow continuation`() = runBlocking {
        (1..5).asFlow().filter { it % 2 == 1 }.collect {
            println(it)
        }
    }

    @Test
    fun `test flow builder`() = runBlocking {
        /**
         * flowof 构建器
         */
        flowOf("one", "two", "three").onEach { delay(1000) }.collect {
            println(it)
        }
    }

    @Test
    fun `test flow context 1`() = runBlocking<Unit> {
        /**
         * 错误的切换上下文方式
         */
        flow {
            withContext(Dispatchers.IO) {
                for (i in 1..5) {
                    delay(1000)
                    emit(i)
                }
            }
        }.collect {
            println(it)
        }
    }

    @Test
    fun `test flow context 2`() = runBlocking<Unit> {
        flow {
            for (i in 1..5) {
                delay(1000)
                emit(i)
            }
        }.flowOn(Dispatchers.IO) //指定流生产的线程
            .collect {
                withContext(Dispatchers.Default) {
                    println(it)
                }
            }
    }

    @Test
    fun `test flow launch`() = runBlocking {
        (0..5).asFlow().onEach { item -> println("item:$item ${Thread.currentThread().name}") }
            .flowOn(Dispatchers.Default) //指定流发起线程
            .launchIn(CoroutineScope(Dispatchers.IO))  //指定收集线程
            .join()
    }

    @Test
    fun `test flow cancel`() = runBlocking {
        /**
         * 超时取消的例子
         */
        try { //会抛出异常
            withTimeout(2500) {
                simpleFlow1().collect {
                    println("collect:$it")
                }
            }
        } catch (e: Exception) {
            println("time out")
        }
        println("Done")
    }

    private fun simpleFlow1() = flow {
        for (i in 0..5) {
            delay(1000)
            println("emit:$i")
            emit(i)
        }
    }


    @Test
    fun `test flow cancel check`() = runBlocking {
//        simpleFlow1().collect{
//            println("collect $it")
//            if (it == 3) {
//                cancel() //取消 会抛出异常
//            }
//        }

        (1..5).asFlow().cancellable().collect {
            println("item:$it")
            if (it == 2) {
                cancel()
            }
        }
    }

    @Test
    fun `test flow backpressure 1`() = runBlocking {
        val time = measureTimeMillis {
            (1..10).asFlow().onEach {
                println("emit $it")
                delay(100)
            }.collect {
                delay(300)
                println("collect $it")
            }
        }
        println("total time:$time") //总计4000
    }

    @Test
    fun `test flow backpressure 2`() = runBlocking {
        val time1 = measureTimeMillis {
            (1..10).asFlow().onEach {
                println("emit $it")
                delay(100)
            }.buffer()  //提供一个缓存缩短运行时间
                .collect {
                    delay(300)
                    println("collect $it")
                }
        }
        println("total time:$time1") //总计3000
    }

    @Test
    fun `test flow backpressure 3`() = runBlocking {
        val time2 = measureTimeMillis {
            (1..10).asFlow().onEach {
                println("emit $it")
                delay(100)
            }.flowOn(Dispatchers.IO)
                .collect {
                    delay(300)
                    println("collect $it")
                }
        }
        println("total time:$time2") //总计3000
    }

    @Test
    fun `test flow backpressure 4`() = runBlocking {
        val time = measureTimeMillis {
            (1..10).asFlow().onEach {
                println("emit $it")
                delay(100)
            }.conflate()   //会忽略来不及处理的元素
                .collect {
                    delay(300)
                    println("collect $it")
                }
        }

        println("total time:$time") //总计1600
    }

    @Test
    fun `test flow backpressure 5`() = runBlocking {
        val time = measureTimeMillis {
            (1..10).asFlow().onEach {
                println("emit $it")
                delay(100)
            }.collectLatest {  //只收集了最后一个
                delay(300)
                println("collect $it")
            }
        }

        println("total time:$time") //总计1300
    }


    @Test
    fun `test flow map`() = runBlocking {
        flow {
            for (i in 1..5) {
                emit(i)
                delay(1000)
            }
        }.map {
            "map $it"
        }.collect {
            println("collect $it")
        }
    }

    @Test
    fun `test flow transform`() = runBlocking {
        flow {
            for (i in 1..5) {
                emit(i)
                delay(1000)
            }
        }.transform {
            emit("transform $it")
            emit(it)
        }.collect {
            println("collect $it, type:${it.javaClass.name}")
        }
    }

    @Test
    fun `test flow limit length operator`() = runBlocking {
        flow {
            for (i in 1..5) {
                emit(i)
            }
        }.take(2).collect {
            println("collect $it")
        }
    }

    @Test
    fun `test flow terminal operator`() = runBlocking {
        val sum =
            (1..5).asFlow().map { it * it }.reduce { accumulator, value -> accumulator + value }
        println("sum:$sum")
    }


    @Test
    fun `test flow zip operator`() = runBlocking {
        val numbers = (1..3).asFlow().onEach { delay(300) }
        val strings = flowOf("one", "two", "three").onEach { delay(200) }
        val startTime = System.currentTimeMillis()
        numbers.zip(strings) { int, str -> "$int,$str" }.collect {
            println("collect $it   start from ${System.currentTimeMillis() - startTime} ms") //两个流是并发的
        }
    }

    @Test
    fun `test flow flatMapConcat`() = runBlocking {
        /**
         * 有情况：一个flow中的每个元素是获取另一个流的参数
         */
        val start = System.currentTimeMillis()
        (1..3).asFlow().onEach { delay(100) }
            .map {
                requestFlow(it)
            }.collect {
                //这里获取的是flow
                println("collect $it  from start time ${System.currentTimeMillis() - start} ms")
            }

        /**
         * 这里将数字流转化为字符串流数组（如上例），然后又将字符串流数组展平变成了一个字符串流
         */
        val start1 = System.currentTimeMillis()
        (1..3).asFlow().onEach { delay(100) }
            .flatMapConcat {
                requestFlow(it)
            }.collect {
                println("collect $it ,from start time ${System.currentTimeMillis() - start1} ms")
            }
    }

    private fun requestFlow(index: Int) = flow<String> {
        emit("first")
        delay(500)
        emit("second")
    }

    @Test
    fun `test flat`() {
        val list = listOf(listOf("1", "2"), listOf("3", "4"), listOf("5", "6"))
        //普通列表展开
        list.flatMap { Iterable { it.iterator() } }.forEach {
            println(it)
        }
        list.flatten().forEach {
            println(it)
        }
    }

    @Test
    fun `test flow flatMapMerge`() = runBlocking {
        val start = System.currentTimeMillis()
        (1..3).asFlow().onEach { delay(100) }
            .flatMapMerge {
                requestFlow(it)
            }.collect {
                println("collect $it ,from start time ${System.currentTimeMillis() - start} ms")
            }
    }

    @Test
    fun `test flow flatMapLatest`() = runBlocking {
        val start = System.currentTimeMillis()
        (1..3).asFlow().onEach { delay(100) }
            .flatMapLatest {//类似collectLatest
                requestFlow(it)
            }.collect {
                println("collect $it ,from start time ${System.currentTimeMillis() - start} ms")
            }
    }

    @Test
    fun `test flow exception`() = runBlocking {
        /**
         * 收集时异常
         */
        try {
            simpleFlow2().collect {
                println("collect $it")
                throw java.lang.NullPointerException()
            }
        } catch (e: Exception) {
            println("exception")
        }
    }

    private fun simpleFlow2() = flow<Int> {
        for (i in 1..3) {
            println("emit $i")
            emit(i)
        }
    }

    @Test
    fun `test flow exception 1`() = runBlocking {
        /**
         * 发射时异常
         */
        try {
            simpleFlow3().catch {
                println("get emit exception") //这里可以获取发射时的异常
            }.collect {
                println("collect $it")
            }
        } catch (e: Exception) {
            println("exception") //这里没有输出
        }
    }

    private fun simpleFlow3() = flow<Int> {
        for (i in 1..3) {
            emit(i)
            delay(100)
            throw NullPointerException()
        }
    }

    @Test
    fun `test flow complete in finally`() = runBlocking {
        try {
            flow {
                repeat(3) {
                    emit(it)
                }
            }.collect {
                println("collect $it")
            }
        } finally {
            println("Done") //这里在collect执行完后执行
        }
    }

    @Test
    fun `test flow complete on onCompletion`() = runBlocking {
        flow {
            repeat(3) {
                println("emit $it")
                emit(it)
                throw java.lang.RuntimeException("test exception")
            }
        }.onCompletion {
            println("completion") //这里也会在收集完后执行,这里还可以拿到异常信息,发射异常和collect异常都能获取
            println(it?.message)
        }.collect {
            println("collect $it")
            throw java.lang.NullPointerException()
        }
    }

    /**
     * channel 协程间通讯的通道,并发安全队列
     */

    @Test
    fun `test channel builder 1`() = runBlocking {
        val channel = Channel<Int>(2)

        val job1 = launch {
            repeat(Int.MAX_VALUE) {
                println("send:$it")
                channel.send(it)
                delay(1000)
            }
        }

        val job2 = launch {
            while (true) {
                println("receive:${channel.receive()}")
                delay(2000)
            }
        }
        joinAll(job1, job2)
    }

    @Test
    fun `test channel builder 2`() = runBlocking {
        val channel = Channel<Int>(Channel.UNLIMITED) //无限大
        val producer = launch {
            repeat(10) {
                println("send $it")
                channel.send(it)
            }
            channel.close() //关闭通道 下游获取数据协程才能退出
        }
        val consumer = launch {
            val iterator = channel.iterator()
            while (iterator.hasNext()) {
                println("receive:${iterator.next()}")
                delay(500)
            }
        }
        joinAll(producer, consumer)
    }

    @Test
    fun `test channel builder 3`() = runBlocking {
        /**
         * 生产者便携封装
         */
        val producerChannel = produce {
            repeat(10) {
                println("send $it")
                send(it)
            }
            close()
        }

        val job = launch {
            val iterator = producerChannel.iterator()
            while (iterator.hasNext()) {
                println("receive:${iterator.next()}")
                delay(1000)
            }
        }
        job.join()
    }

    @Test
    fun `test channel builder 4`() = runBlocking {
        /**
         * 消费者便携封装
         */
        val consumerChannel = actor<Int> {
            val iterator = iterator()
            while (iterator.hasNext()) {
                println("receive: ${iterator.next()}")
            }
        }

        val job = launch {
            repeat(10) {
                delay(500)
                consumerChannel.send(it)
            }
            consumerChannel.close()  //关闭
        }
        job.join()
    }

    @Test
    fun `test channel close`() = runBlocking {
        val channel = Channel<Int>(capacity = 2)

        val sender = launch {
            repeat(5) {
                println("send: $it")
                channel.send(it)
                delay(200)
            }
            channel.close()
            println("send close:${channel.isClosedForSend} \nreceive close:${channel.isClosedForReceive}")
        }

        val receiver = launch {
            val iterator = channel.iterator()
            while (iterator.hasNext()) {
                println("receive: ${iterator.next()}")
                delay(1000)
            }
            println("send close:${channel.isClosedForSend} \nreceive close:${channel.isClosedForReceive}")
        }
        joinAll(sender, receiver)
    }

    @Test
    fun `test broadcast channel`() = runBlocking<Unit> {
        /**
         * 普通channel是一对一的发送接收
         * 广播资方是一对多了
         */

        val channel = BroadcastChannel<Int>(Channel.BUFFERED)

        val producer = GlobalScope.launch {
            val start = System.currentTimeMillis()
            repeat(5) {
                delay(2000)
                println("send $it,at ${System.currentTimeMillis() - start}")
                channel.send(it)
            }
            channel.close()
        }

        List(3) {
            GlobalScope.launch {
                val receiveChannel = channel.openSubscription()
                for (i in receiveChannel) {
                    println("[#$it] receive $i")
                }
            }
        }

        producer.join()
    }

    private suspend fun CoroutineScope.getUserFromLocal() = async(Dispatchers.IO) {
        delay(1400)
        "local"
    }

    private suspend fun CoroutineScope.getUserFromRemote() = async(Dispatchers.IO) {
        delay(1200)
        "remote"
    }

    @Test
    fun `test select await 1`() = runBlocking<Unit> {
        val job1 = launch {
            val result1 = getUserFromLocal()
            val result2 = getUserFromRemote()

            val realresult = select<String> {
                result1.onAwait { it }
                result2.onAwait { it }
            }
            println("result:$realresult")
        }

        job1.join()
    }

    @Test
    fun `test select await 2`() = runBlocking<Unit> {
        val result = select {
            async(Dispatchers.IO) {
                delay(2000)
                "2000"
            }.onAwait { it }

            async {
                delay(1000)
                "30000"
            }.onAwait { it }
        }

        println(result)
    }

    @Test
    fun `test select channel`() = runBlocking<Unit> {
        val channel1 = Channel<Int>()
        val channel2 = Channel<String>()
        val sender1 = GlobalScope.launch {
            delay(1000)
            channel1.send(1000)
        }
        val sender2 = GlobalScope.launch {
            delay(800)
            channel2.send("800")
        }

        val result = select<Any> {
            channel1.onReceive { it }
            channel2.onReceive { it }
        }
        println(result)
//        joinAll(sender1, sender2)
    }

    @Test
    fun `test select clause0`() = runBlocking<Unit> {
        val job1 = GlobalScope.launch {
            delay(1000)
            println("run job 1")
        }
        val job2 = GlobalScope.launch {
            delay(200)
            println("run job 2")
        }

        val result = select<Unit> {
            job1.onJoin { println("select job 1") }
            job2.onJoin { println("select job 2") }
        }
        delay(2000)
    }

    @Test
    fun `test select clause2`() = runBlocking {
        val channels = listOf(Channel<Int>(), Channel<Int>())
        println(channels)

        /**
         * 意思是 两个协程分别像两个channel中发送数据，谁发送得快 谁就发？
         */
        launch(Dispatchers.IO) {
            select<Unit?> {
                launch {
                    delay(10)
                    channels[0].onSend(100) {
                        println("sent on $it")
                    }
                }
                launch {
                    delay(100)
                    channels[1].onSend(200) {
                        println("sent on $it")
                    }
                }
            }
        }

        GlobalScope.launch {
            println(channels[0].receive())
        }

        GlobalScope.launch {
            println(channels[1].receive())
        }

        delay(1000)
    }

    @Test
    fun `test select flow`() = runBlocking {
        coroutineScope {
            listOf(getUserFromLocal(), getUserFromRemote()).asFlow()
                .map { it.await() }
                .collect {
                    println(it)
                }
        }
    }

    @Test
    fun `test concurrent safe 1`() = runBlocking {
        var count = AtomicInteger(0)
        var count1 = 0
        repeat(1000) {
            GlobalScope.launch {
                count.incrementAndGet()
                count1++
            }
        }
        println("count:$count")
        println("count1:$count1")
    }

    @Test
    fun `test concurrent safe 2`() = runBlocking {
        val channel = Channel<Int>()
        repeat(1000) {
            GlobalScope.launch {
                channel.send(1)
            }
        }
        channel.close()

        var count = 0
        for (i in channel) {
            count += i
        }
        println("count:$count")
    }

    @Test
    fun `test concurrent safe 3`() = runBlocking {
        /**
         * 加锁
         */
        var count = 0
        val mutex = Mutex()
        List(1000) {
            GlobalScope.launch {
                mutex.withLock {
                    count++
                }
            }
        }.joinAll()

        println("count:$count")
    }

    @Test
    fun `test concurrent safe 4`() = runBlocking {
        /**
         * 信号量
         */
        var count = 0
        val semaphore = Semaphore(1)
        List(1000) {
            GlobalScope.launch {
//                semaphore.acquire()
//                count++
//                semaphore.release()

                semaphore.withPermit {
                    count++
                }
            }
        }.joinAll()
        println("count:$count")
    }

    @Test
    fun `test tips`() = runBlocking {
        var count = 0
        count = List(1000) {
            GlobalScope.async { 1 }
        }.map { it.await() }.sum()
        println("count:$count")
    }
}













